#include <cassert>
#include <google/protobuf/text_format.h>
#include <google/protobuf/io/zero_copy_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include "operations.pb.h"
#include "Consumer.h"
#include "ScanOperationNode.h"
#include "ComputeOperationNode.h"
#include "FilterOperationNode.h"
#include "GroupByOperationNode.h"
#include "UnaryExpressionNode.h"
#include "BinaryExpressionNode.h"
#include "IfExpressionNode.h"
#include "ColumnExpressionNode.h"
#include "ConstantExpressionNode.h"
#include "AggregationData.h"
#include "SharedArray.h"
#include "operators.h"
#include "types.h"
#include "Parser.h"

using namespace google::protobuf;
using namespace google::protobuf::io;

static OperationNode* parseOperation(const Operation& op, Server* server);

static ScanOperationNode* parseScanOperation(const ScanOperation& op, Server* server) {
	assert(op.column_size() == op.type_size());

	SharedArray<Type> types(op.column_size());
	SharedArray<int> columns(op.column_size());

	for (int i = 0; i < op.column_size(); ++i) {
		types[i] = static_cast<Type>(op.type(i));
		columns[i] = op.column(i);
	}

	return new ScanOperationNode(types, columns, server);
}

static ExpressionNode* parseExpression(const Expression& expr) {
	switch (expr.operator_()) {
		case Expression::CONSTANT:
			assert(expr.children_size() == 0);
			if (expr.has_constant_int32()) {
				return new ConstantExpressionNode<int32_t>(expr.constant_int32());
			} else if (expr.has_constant_double()) {
				return new ConstantExpressionNode<double>(expr.constant_double());
			} else {
				assert(expr.has_constant_bool());
				return new ConstantExpressionNode<bool>(expr.constant_bool());
			}

		case Expression::COLUMN:
			assert(expr.children_size() == 0);
			assert(expr.has_column_id());
			return new ColumnExpressionNode(expr.column_id());

		case Expression::IF:
			assert(expr.children_size() == 3);
			return new IfExpressionNode(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)), parseExpression(expr.children(2)));

		case Expression::ADD:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<AddOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::SUBTRACT:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<SubtractOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::MULTIPLY:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<MultiplyOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::FLOATING_DIVIDE:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<FloatingDivideOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::LOG:
			assert(expr.children_size() == 1);
			return new UnaryExpressionNode<LogOperator>(parseExpression(expr.children(0)));

		case Expression::NEGATE:
			assert(expr.children_size() == 1);
			return new UnaryExpressionNode<NegateOperator>(parseExpression(expr.children(0)));

		case Expression::LOWER:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<LowerOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::GREATER:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<GreaterOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::EQUAL:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<EqualOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::NOT_EQUAL:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<NotEqualOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::NOT:
			assert(expr.children_size() == 1);
			return new UnaryExpressionNode<NotOperator>(parseExpression(expr.children(0)));

		case Expression::OR:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<OrOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		case Expression::AND:
			assert(expr.children_size() == 2);
			return new BinaryExpressionNode<AndOperator>(parseExpression(expr.children(0)),
					parseExpression(expr.children(1)));

		default:
			assert(false);
	}

	return 0;
}

static ComputeOperationNode* parseComputeOperation(const ComputeOperation& op, Server* server) {
	OperationNode* source = parseOperation(op.source(), server);
	SharedArray<ExpressionNode*> expressionRoots(op.expressions_size());

	for (int i = 0; i < op.expressions_size(); ++i) {
		expressionRoots[i] = parseExpression(op.expressions(i));
	}

	return new ComputeOperationNode(source, expressionRoots);
}

static FilterOperationNode* parseFilterOperation(const FilterOperation& op, Server* server) {
	OperationNode* source = parseOperation(op.source(), server);
	ExpressionNode* expression = parseExpression(op.expression());

	return new FilterOperationNode(source, expression);
}

static AggregationData parseAggregation(const Aggregation& agg) {
	assert(agg.has_aggregated_column() || agg.type() == Aggregation::COUNT);

	int column = agg.has_aggregated_column() ? agg.aggregated_column() : -1;
	return AggregationData(static_cast<AggregationData::AggregationType>(agg.type()), column);
}

static GroupByOperationNode* parseGroupByOperation(const GroupByOperation& op, Server* server) {
	OperationNode* source = parseOperation(op.source(), server);

	SharedArray<int> groupedColumns(op.group_by_column_size());
	for (int i = 0; i < op.group_by_column_size(); ++i) {
		groupedColumns[i] = op.group_by_column(i);
	}

	SharedArray<AggregationData> aggregations(op.aggregations_size());
	for (int i = 0; i < op.aggregations_size(); ++i) {
		aggregations[i] = parseAggregation(op.aggregations(i));
	}

	return new GroupByOperationNode(source, groupedColumns, aggregations);
}

OperationNode* parseOperation(const Operation& op, Server* server) {
	if (op.has_scan()) {
		return parseScanOperation(op.scan(), server);
	} else if (op.has_compute()) {
		return parseComputeOperation(op.compute(), server);
	} else if (op.has_filter()) {
		return parseFilterOperation(op.filter(), server);
	} else {
		assert(op.has_group_by());
		return parseGroupByOperation(op.group_by(), server);
	}
}

Consumer* Parser::parseOperationsTree(int inputFileDescriptor, Server* server) {
	Operation op;
	FileInputStream input(inputFileDescriptor);
	if (google::protobuf::TextFormat::Parse(&input, &op) == 0)
		return 0;

	OperationNode* root = parseOperation(op, server);

	return new Consumer(root, server);
}
